package com.zetyun.tiger.connector.op.v2;

import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.collect.Sets;
import com.zetyun.rt.meta.annotation.ActionMeta;
import com.zetyun.rt.sdk.action.MapAction;
import com.zetyun.rt.sdk.model.RtEvent;
import com.zetyun.rt.sdk.operator.OperatorContext;
import com.zetyun.tiger.connector.cdc.SimpleRedisContainer;
import com.zetyun.tiger.connector.util.SpecialDateFormatUtil;
import lombok.SneakyThrows;
import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisConfigBase;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisSentinelConfig;
import redis.clients.jedis.JedisPoolConfig;
import redis.clients.jedis.JedisSentinelPool;

import java.util.Objects;
import java.util.Set;

@ActionMeta(
        id = "tiger.redis-opv2",
        tags = {"redis"},
        category = "map",
        name = "RedisOpV2",
        description = "Redis V2操作算子",
        parameterClass = RedisOpV2Parameter.class,
        status = ActionMeta.Status.PUBLIC
)
public class RedisOpV2Action extends MapAction {


    private SimpleRedisContainer rc;
    private RedisOpV2Parameter parameters;
    private ObjectMapper om;

    // 客户经理
    private String conntrPrefix;

    private String custCntPrefix;

    // 公共揽存
    private String commonPrefix;

    private String commonCntPrefix;

    private final Integer ttl = 60 * 60 * 24;  // 一天

    @Override
    public void init(OperatorContext context) {
        super.init(context);

        om = new ObjectMapper();
        parameters = context.getParameters();
        conntrPrefix = parameters.getConntrPrefix();
        custCntPrefix = parameters.getCustCntPrefix();
        commonPrefix = parameters.getCommonPrefix();
        commonCntPrefix = parameters.getCommonCntPrefix();
        FlinkJedisSentinelConfig conf =
                new FlinkJedisSentinelConfig.Builder()
                        .setMasterName(parameters.getMasterName())
                        .setSentinels(parameters.getSentinels())
                        .setPassword(parameters.getPassword())
                        .setDatabase(parameters.getDatabase())
                        .build();
        rc = build(conf);
        try {
            rc.open();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }


    @Override
    public RtEvent apply(RtEvent in) throws Exception {
        // hash
        // key:客户经理号
        //          field: 客户号
        //          value: Set<内部账号>
        String dayId = in.getField(parameters.getDayId(), String.class);
        String opacInstnNo = in.getField(parameters.getOpacInstnNo(), String.class);
        String conntrNo = in.getField(parameters.getConntrNo(), String.class);
        String custNo = in.getField(parameters.getCustNo(), String.class);
        String acctNo = in.getField(parameters.getAcctNo(), String.class);
        String opacType = in.getField(parameters.getOpacType(), String.class);

        String acctNoCacheName = conntrPrefix.concat(conntrNo);
        String dayIdCacheName = custCntPrefix.concat(dayId);


        Set<String> allAcctNos = Sets.newHashSet();
//        strToSet(rc.hget(acctNoCacheName, custNo));

        /*
        目的：
        1. 统计账户开销户明细，找出开销户数据
        2. 统计客户数

        处理流程：
        1. 开户
            从SAACNACN中来到一个活存账户，首先获取它的客户号，判断是否在客户经理总缓存中
           如果不存在，则为新增客户，并且为新增开户，将此账户加入缓存中，并且缓存中客户数 +1 ：数据放出
           如果存在：①，获取客户对应的所有账号集合，判断活存账户是否在集合中，
                    如果存在：则 不是新增客户，也不是新增开户
                    如果不存在：则 不是新增客户， 是新增开户。 将此账户加入缓存中，缓存中客户数不变 ：数据放出
        2. 销户
            从SAACNACN中来到一个活存账户，首先获取它的客户号，判断是否在客户经理总缓存中
            如果不存在，则为流失客户，并且为销户 （其实是脏数据） todo  缓存中客户数 -1  ：数据放出 有待考究
            如果存在：①，获取客户对应的所有账号集合，判断活存账户是否在集合中，并且判断集合大小
                    如果集合大小 = 1 {
                    如果存在：则 是流失客户，并且是销户    todo   将此账户从缓存中删除，缓存中客户数减一 ：数据放出 暂无异议
                    如果不存在：则 不是流失客户， 不是销户。 （其实是脏数据） todo   缓存中客户数不变 ：数据放出 有待考究
                    }

                    如果集合大小 ！= 1 {
                    如果存在：则 不是流失客户，并且是销户  todo   将此账户从缓存中删除，缓存中客户数不变 ：数据放出 暂无异议
                    如果不存在：则 不是流失客户， 不是销户。 （其实是脏数据） todo   缓存中客户数不变 ：数据放出 有待考究
                    }

         */

        if (opacType.equals("01")) {
            // 开户
            // 不存在客户经理
            if (conntrNo.isEmpty()) {
                // 公共揽存数据
                String opacAcctCacheName = commonPrefix.concat(opacInstnNo);
                String opacDayIdCacheName = commonCntPrefix.concat(dayId);
                //
                Boolean hexists = rc.hexists(opacAcctCacheName, custNo);
                if (!hexists) {
                    allAcctNos.add(acctNo);
                    rc.hset(opacAcctCacheName, custNo, setToStr(allAcctNos), null);   // 设置一天的过期时间
                    // TODO: 2022/09/15 客户数 +1 ：数据放出
                    incrCustCnt(opacDayIdCacheName,opacInstnNo);
                } else {
                    allAcctNos = strToSet(rc.hget(opacAcctCacheName, custNo));
                    boolean contains = allAcctNos.contains(acctNo);
                    if (!contains) {
                        allAcctNos.add(acctNo);
                        rc.hset(opacAcctCacheName, custNo, setToStr(allAcctNos), null);   // 设置一天的过期时间
                        // TODO: 2022/09/15 数据放出
                    } else {
//                    return null;
                    }
                }
            } else {
                // 存在客户经理
                Boolean hexists = rc.hexists(acctNoCacheName, custNo);
                if (!hexists) {
                    allAcctNos.add(acctNo);
                    rc.hset(acctNoCacheName, custNo, setToStr(allAcctNos), null);   // 设置一天的过期时间
                    // TODO: 2022/09/15 客户数 +1 ：数据放出
                    incrCustCnt(dayIdCacheName,conntrNo);
                } else {
                    allAcctNos = strToSet(rc.hget(acctNoCacheName, custNo));
                    boolean contains = allAcctNos.contains(acctNo);
                    if (!contains) {
                        allAcctNos.add(acctNo);
                        rc.hset(acctNoCacheName, custNo, setToStr(allAcctNos), null);   // 设置一天的过期时间
                        // TODO: 2022/09/15 数据放出
                    } else {
//                    return null;
                    }
                }
            }


        } else {
            // 销户


            if (conntrNo.isEmpty()) {
                // 不存在客户经理
                // 公共揽存数据
                String opacAcctCacheName = commonPrefix.concat(opacInstnNo);
                String opacDayIdCacheName = commonCntPrefix.concat(dayId);
                Boolean hexists = rc.hexists(opacAcctCacheName, custNo);
                if (!hexists) {
                    // TODO: 2022/09/15 缓存中客户数 -1  ：数据放出 有待考究
                    decrCustCnt(opacDayIdCacheName,opacInstnNo);
                } else {
                    allAcctNos = strToSet(rc.hget(opacAcctCacheName, custNo));
                    int size = allAcctNos.size();
                    boolean contains = allAcctNos.contains(acctNo);
                    if (size == 1) {
                        if (contains) {
//                        allAcctNos.remove(acctNo);
//                        rc.hset(acctNoCacheName, custNo, setToStr(allAcctNos), null);   // 设置一天的过期时间
                            rc.hdel(opacAcctCacheName, custNo);
                            // TODO: 2022/09/15 缓存中客户数减一 ：数据放出
                            decrCustCnt(opacDayIdCacheName,opacInstnNo);
                        } else {
                            // TODO: 2022/09/15  缓存中客户数不变 ：数据放出 有待考究
//                        return null;
                        }
                    } else {
                        if (contains) {
                            allAcctNos.remove(acctNo);
                            rc.hset(opacAcctCacheName, custNo, setToStr(allAcctNos), null);   // 设置一天的过期时间
                            // TODO: 2022/09/15 缓存中客户数不变 ：数据放出 暂无异议
                        } else {
                            // TODO: 2022/09/15  缓存中客户数不变 ：数据放出 有待考究
//                        return null;
                        }
                    }
                }
            } else {
                // 存在客户经理
                Boolean hexists = rc.hexists(acctNoCacheName, custNo);
                if (!hexists) {
                    // TODO: 2022/09/15 缓存中客户数 -1  ：数据放出 有待考究
                    decrCustCnt(dayIdCacheName,conntrNo);
                } else {
                    allAcctNos = strToSet(rc.hget(acctNoCacheName, custNo));
                    int size = allAcctNos.size();
                    boolean contains = allAcctNos.contains(acctNo);
                    if (size == 1) {
                        if (contains) {
//                        allAcctNos.remove(acctNo);
//                        rc.hset(acctNoCacheName, custNo, setToStr(allAcctNos), null);   // 设置一天的过期时间
                            rc.hdel(acctNoCacheName, custNo);
                            // TODO: 2022/09/15 缓存中客户数减一 ：数据放出
                            decrCustCnt(dayIdCacheName,conntrNo);
                        } else {
                            // TODO: 2022/09/15  缓存中客户数不变 ：数据放出 有待考究
//                        return null;
                        }
                    } else {
                        if (contains) {
                            allAcctNos.remove(acctNo);
                            rc.hset(acctNoCacheName, custNo, setToStr(allAcctNos), null);   // 设置一天的过期时间
                            // TODO: 2022/09/15 缓存中客户数不变 ：数据放出 暂无异议
                        } else {
                            // TODO: 2022/09/15  缓存中客户数不变 ：数据放出 有待考究
//                        return null;
                        }
                    }
                }
            }



//            if (allAcctNos.size() == 1 && allAcctNos.contains(custNo)) {
//                // 表明账号信息里面存的是当前销户账户，并且只有一条数据，则为销户数据
//                rc.hdel(dayIdCacheName, custNo);
//            }
//            allAcctNos.remove(custNo);
//            rc.hset(acctNoCacheName, custNo, setToStr(allAcctNos), null);
        }

        // 获取当前客户经理的今日客户数
        long customerCnt = getCustCnt(dayIdCacheName,conntrNo);

        // 获取当前处理时间
        long tms = System.currentTimeMillis();
        String ymdHms = SpecialDateFormatUtil.toYmdHms(tms);
        in.setValue("RT_CUST_CNT", customerCnt);
        in.setValue("RT_PROC_TMS", tms);
        in.setValue("RT_PROC_TIME", ymdHms);

        return in;
    }


    @Override
    public void close() throws Exception {
        super.close();
        rc.close();
    }




    private Long incrCustCnt(String cacheName, String conntrNo) {
        long res = 1;
        Boolean hexists = rc.hexists(cacheName, conntrNo);
        if (!hexists) {
            rc.hset(cacheName, conntrNo, "1", ttl);
        } else {
            String strVal = rc.hget(cacheName, conntrNo);
            long cacheValue = Long.parseLong(strVal);
            res = cacheValue + 1;
            rc.hset(cacheName, conntrNo, res + "", ttl);
        }
        return res;
    }

    private Long decrCustCnt(String cacheName, String conntrNo) {
        long res = -1;
        Boolean hexists = rc.hexists(cacheName, conntrNo);
        if (!hexists) {
            rc.hset(cacheName, conntrNo, "-1", ttl);
        } else {
            String strVal = rc.hget(cacheName, conntrNo);
            long cacheValue = Long.parseLong(strVal);
            res = cacheValue - 1;
            rc.hset(cacheName, conntrNo, res + "", ttl);
        }
        return res;
    }

    private Long getCustCnt(String cacheName, String conntrNo) {
        long res = 0;
        Boolean hexists = rc.hexists(cacheName, conntrNo);
        if (hexists) {
            String strVal = rc.hget(cacheName, conntrNo);
            res = Long.parseLong(strVal);
        }
        return res;
    }


    @SneakyThrows
    private Set<String> strToSet(String jsonStr) {
        if (jsonStr == null || jsonStr.isEmpty()) {
            return Sets.newHashSet();
        }
        return om.readValue(jsonStr, new TypeReference<Set<String>>() {
        });
    }

    @SneakyThrows
    private String setToStr(Set<String> set) {
        return om.writeValueAsString(set);
    }


    public static SimpleRedisContainer build(FlinkJedisSentinelConfig jedisSentinelConfig) {
        Objects.requireNonNull(jedisSentinelConfig, "Redis sentinel config should not be Null");

        GenericObjectPoolConfig genericObjectPoolConfig = getGenericObjectPoolConfig(jedisSentinelConfig);

        JedisSentinelPool jedisSentinelPool = new JedisSentinelPool(jedisSentinelConfig.getMasterName(),
                jedisSentinelConfig.getSentinels(), genericObjectPoolConfig,
                jedisSentinelConfig.getConnectionTimeout(), jedisSentinelConfig.getSoTimeout(),
                jedisSentinelConfig.getPassword(), jedisSentinelConfig.getDatabase());
        return new SimpleRedisContainer(jedisSentinelPool);
    }

    private static GenericObjectPoolConfig getGenericObjectPoolConfig(FlinkJedisConfigBase jedisConfig) {
        GenericObjectPoolConfig genericObjectPoolConfig = jedisConfig.getTestWhileIdle() ? new JedisPoolConfig() : new GenericObjectPoolConfig();
        genericObjectPoolConfig.setMaxIdle(jedisConfig.getMaxIdle());
        genericObjectPoolConfig.setMaxTotal(jedisConfig.getMaxTotal());
        genericObjectPoolConfig.setMinIdle(jedisConfig.getMinIdle());
        genericObjectPoolConfig.setTestOnBorrow(jedisConfig.getTestOnBorrow());
        genericObjectPoolConfig.setTestOnReturn(jedisConfig.getTestOnReturn());

        return genericObjectPoolConfig;
    }
}
